package com.yanxu;

import com.yanxu.domain.Event;
import com.yanxu.source.CustomSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025/7/27-2025
 * @Description: 时间语义
 *  水位线相关
 */
public class Api14_WaterMarkSample {
    public static void main(String[] args) throws Exception {
        //定义执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Event> source = env.addSource(new CustomSource());
        /*SingleOutputStreamOperator<String> map = source.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event event) throws Exception {
                return event.getName();
            }
        });*/

        //添加水位线机制
        SingleOutputStreamOperator<Event> operator = source.assignTimestampsAndWatermarks(
                //定义水位线延迟策略，针对有可能出现的无序情况，将延迟时间定义为 5 秒钟。
                WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        // 定义时间戳分配器。我们的一个 业务数据，比如 Event 中。
                        // 可能包含很多时间，比如 创建时间、更新时间、修改时间 .....等等。
                        // 因此，我们需要指定到底需要哪个时间戳字段进行判断，进行统计啊。
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long l) {
                                return element.getTimestamp();
                            }
                        })
        );

        //打印
        operator.print(">>> ");

        //启动执行环境
        env.execute();
    }
}